package com.youbu.bigdata.utils;

import com.youbu.bigdata.bean.SparkOptionConfig;
import com.youbu.bigdata.bean.YbSparkConfig;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;

/**
 *有步Spark开发类
 * @author sunfangwei
 **/
public class YbSpark {

	private SparkSession sparkSession;

	/**
	 * 构造方法，自定义sparksession
	 * @param sparkSession
	 */
	public YbSpark(SparkSession sparkSession){
		this.sparkSession = sparkSession;
	}

	/**
	 * 构造方法，自定义SparkConfig
	 * @param conf
	 */
	public YbSpark(SparkConf conf){
		this.sparkSession = SparkUtil.getSparkSession(conf);
	}

	/**
	 * 构造方法
	 * @param ybSparkConfig
	 */
	public YbSpark(YbSparkConfig ybSparkConfig){
		this.sparkSession = SparkUtil.getSparkSession(ybSparkConfig.getSparkConf());
	}

	/**
	 * 获取文件内容并转换为JAVARDD
	 * @param path 文本路径
	 * @return
	 */
	public JavaRDD<String> getTextFile(String path){
		return this.sparkSession.read().textFile(path).javaRDD().persist(StorageLevel.OFF_HEAP());
	}

	/**
	 * 保存PairRDD为Text文件
	 * @param pairRDD
	 * @param savePath 保存路径
	 */
	public void saveRddAsTextFile(JavaPairRDD<?, ?> pairRDD, String savePath){
		pairRDD.saveAsTextFile(savePath);
	}

	/**
	 * 保存Rdd为Text文件
	 * @param rdd
	 * @param savePath 保存路径
	 */
	public void saveRddAsTextFile(JavaRDD<?> rdd, String savePath){
		rdd.saveAsTextFile(savePath);
	}

	/**
	 *将Dataset保存入数据库
	 * @param dataset
	 * @param optionConfig
	 */
	public void saveDatasetAsJDBC(Dataset<?> dataset,SparkOptionConfig optionConfig){
		dataset.write().format("jdbc").mode(SaveMode.Append).options(optionConfig.getOptionMap()).save();
	}

	/**
	 * 将Dataset保存为文本文件
	 * @param dataset
	 * @param path
	 */
	public void saveDataSetAsTextFile(Dataset<?> dataset,String path){
		dataset.write().save(path);
	}

	/**
	 * 将Dataset保存为json文件
	 * @param dataset
	 * @param path
	 */
	public void saveDatasetAsJson(Dataset<?> dataset,String path){
		dataset.write().json(path);
	}


	/**
	 * 获取数据表内容
	 * @param optionConfig JDBC连接配置
	 * @return
	 */
	public Dataset<Row> getJDBCDataSet(SparkOptionConfig optionConfig){
		return this.sparkSession.read().format("jdbc").options(optionConfig.getOptionMap()).load().persist(StorageLevel.OFF_HEAP());
	}

	/**
	 * 获取Json内容
	 * @param path
	 * @return
	 */
	public Dataset<Row> getJsonDataSet(String path){
		return this.sparkSession.read().json(path).persist(StorageLevel.OFF_HEAP());
	}

	/**
	 * 关闭SparkSession
	 */
	public void close(){
		this.sparkSession.close();
	}

	/**
	 * 获取sparkSession.
	 */
	public SparkSession getSparkSession() {
		return sparkSession;
	}

	/**
	 * 设置 sparkSession.
	 */
	public void setSparkSession(SparkSession sparkSession) {
		this.sparkSession = sparkSession;
	}
}
